package com.example.springbatchdemo.job;

import com.example.springbatchdemo.Person;
import com.example.springbatchdemo.job.itemProcessor.MyItemProcessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileParseException;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import javax.sql.DataSource;
import java.io.FileNotFoundException;


/**
 * job配置
 */
@Configuration
@EnableBatchProcessing
@Slf4j
public class JobConfig extends DefaultBatchConfigurer {
        @Autowired
        StepBuilderFactory stepBuilderFactory;
        @Autowired
        JobBuilderFactory jobBuilderFactory;


        /**
         * job
         * @param listener
         * @param step1
         * @return
         */
        @Bean
        public Job importPersonJob( JobListener listener,Step step1) {
                return jobBuilderFactory.get("importPersonJob")
                        .incrementer(new RunIdIncrementer())
                        .listener(listener)
                        .flow(step1)
                        .end()
                        .build();
        }
        /**
         * step
         * @param writer
         * @return
         */
        @Bean
        public Step step1(JdbcBatchItemWriter<Person> writer) {
                return stepBuilderFactory.get("step1")
                        .<Person, Person> chunk(2)
                        .reader(reader())
                        .processor(processor())
                        .writer(writer)
                        .faultTolerant()
                        .skipLimit(10)
                        .skip(FlatFileParseException.class)
                        .noSkip(FileNotFoundException.class)
                        .build();
        }
        /**
         * read
         * @return
         */
        @Bean
        public FlatFileItemReader<Person> reader(){
                return new FlatFileItemReaderBuilder<Person>()
                        .name("personItemReader")
                        .resource(new ClassPathResource("sample-data.csv"))
                        .delimited()
                        .names(new String[]{"name", "age"})
                        .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                                setTargetType(Person.class);
                        }})
                        .build();
        }
        /**
         * processor
         * @return
         */
        @Bean
        public MyItemProcessor processor(){
                return new MyItemProcessor();
        }

        /**
         * write
         * @param dataSource
         * @return
         */
        @Bean
        public JdbcBatchItemWriter<Person> write(DataSource dataSource){
                return new JdbcBatchItemWriterBuilder<Person>()
                        .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
                        .sql("INSERT INTO person (name, age) VALUES (:name, :age)")
                        .dataSource(dataSource)
                        .build();
        }



}
